package com.bw.gmall.realtime.app.ods;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.*;

public class BaseLogAppShopOds {
    public static void main(String[] args) throws FileNotFoundException, InterruptedException {
        Long[]   arrTs={1730307600000l, 1730311200000l, 1730314800000l, 1730318400000l, 1730321400000l, 1730324400000l, 1730328400000l, 1730331400000l, 1730334400000l, 1730338400000l, 1730341400000l, 1730344400000l, 1730348400000l, 1730351400000l,
                1730354400000l, 1730358400000l, 1730361400000l, 1730364400000l, 1730368400000l, 1730371800000l, 1730374100000l, 1730378400000l, 1730381800000l, 1730384800000l, 1730388800000l
        };
        // 处理每行数据
          /*              访问店铺行为包含观看店铺自播直播间、观看自制全屏页短视频3秒及以上、
                浏览店铺自制图文3秒及以上、浏览全屏微详情、访问宝贝详情页及店铺其他页面*/
        ArrayList<String> list = new ArrayList<>();
        list.add("o"); //其他有行为
        list.add("a");  //观看店铺自播直播间
        list.add("b");//观看自制全屏页短视频3秒及以上
        list.add("c");//浏览店铺自制图文3秒及以上
        list.add("d");//浏览全屏微详情
        list.add("e");//访问宝贝详情页

        for (Long arrT : arrTs) {
            mockDatas(list,arrT);
            System.out.println("---------------------------------->");
            Thread.sleep(3000);
        }

//1  1730307600000l
//2  1730311200000l
//3  1730314800000l
//4  1730318400000l
//5  1730321400000l
//6  1730324400000l
//7  1730328400000l
//8  1730331400000l
//9  1730334400000l
//9  1730444400000l
//9  1730544400000l
//9  1730644400000l

    }

    /*
     * 每个店铺商品id是不同的
     * */
    public  static  void  mockDatas(ArrayList<String> list, Long ts) throws FileNotFoundException {
        ArrayList<JSONObject> jsonObjects = new ArrayList<>();
        File file = new File("datas/app.2024-10-30.log");
        Scanner scanner = new Scanner(file);
        while (scanner.hasNextLine()) {
            String line = scanner.nextLine();

            JSONObject jsonObj = JSON.parseObject(line);
            JSONObject common = jsonObj.getJSONObject("common");
            JSONObject page = jsonObj.getJSONObject("page");
            common.put("shop_id",new Random().nextInt(10)+1 );
            common.put("mid",new Random().nextInt(100000)+1 );
            int i = new Random().nextInt(2);
            int i1 = new Random().nextInt(5);

            if(page!=null){
                page.put("page_action",list.get(i1));
            }
            common.put("device_type",(i==0?0:1));
            //  1:    2   3   4
            jsonObj.put("ts",ts);

            jsonObjects.add(jsonObj);
        }

        Map<String, Object> config = new HashMap<>();
        config.put("bootstrap.servers", "hadoop102:9092");
        config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka生产者
        Producer<String, String> producer = new KafkaProducer<>(config);

        // 创建JSON对象
        ObjectMapper mapper = new ObjectMapper();
        try {
            // 假设我们有一个JSON字符串
            for (JSONObject jsonObject : jsonObjects) {
                // 发送到Kafka
                producer.send(new ProducerRecord<>("ods_traffic", jsonObject.toJSONString()));
            }




        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }

        scanner.close();



    }
}
